import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IList;
import com.hazelcast.core.IMap;
import com.hazelcast.core.ISemaphore;
import com.hazelcast.core.TransactionalMap;
import com.hazelcast.transaction.TransactionContext;
import com.hazelcast.transaction.TransactionOptions;
import com.hazelcast.transaction.TransactionOptions.TransactionType;
import com.hazelcast.util.ExceptionUtil;

import util.NearCacheTest;
import util.SpringCache;

/**
 * transaction map 开始后，读取相关要素的线程将被阻塞
 * <p>
 * 使用local cache 和ISemaphore强制刷新缓存功能；
 * <p>
 * 有ACK响应确认，确认结束事务的过程，没有读到旧数据的问题
 * 
 * @author wxu
 *
 */
public class TransactionalJCache {

	private static final Logger logger = LoggerFactory.getLogger(TransactionalJCache.class);

	static Runnable tx = new Runnable() {
		@Override
		public void run() {

			try {
				logger.info("start transaction...");
				TransactionOptions options = new TransactionOptions();
				options.setTransactionType(TransactionType.TWO_PHASE);
				options.setTimeout(100000, TimeUnit.MILLISECONDS);
				TransactionContext txCxt = hz.newTransactionContext(options);
				txCxt.beginTransaction();
				TransactionalMap<Object, Object> map = txCxt.getMap(NearCacheTest.MAP);

				try {
					logger.info("before put map(1)");
					Thread.sleep(3000);
					String tmp = String.valueOf(System.currentTimeMillis());
					map.put("1", tmp); // reader1 is blocked
					logger.info("after put map(1)");
					Thread.sleep(3000);
					tmp = String.valueOf(System.currentTimeMillis());
					map.put("2", tmp); // reader2 is blocked
					logger.info("after put map(2)");
					Thread.sleep(5000);

					// evict all client map cache
					IList<Object> evictList = hz.getList(NearCacheTest.SEMAPHORE_EVICT);
					evictList.add("1");
					evictList.add("2");

					int size = hz.getClientService().getConnectedClients().size();
					ISemaphore semaphore = hz.getSemaphore(NearCacheTest.SEMAPHORE_EVICT);
					semaphore.release(size);
					logger.info("client count:{}", size);

					// wait to response from clients
					while (semaphore.tryAcquire()) {
					}

					txCxt.commitTransaction();
					logger.info("transaction committed");
				} catch (RuntimeException t) {
					txCxt.rollbackTransaction();
					throw t;
				}

				Thread.sleep(5000);
			} catch (Exception e) {
				logger.error("error:{}", ExceptionUtil.toString(e));
			} finally {
				// Hazelcast.shutdownAll();
				logger.info("system exit.");
				System.exit(0);
			}

		}
	};

	private static HazelcastInstance hz;

	public static void main(String[] args) throws Exception {

		hz = Hazelcast.newHazelcastInstance();
		IMap<String, String> testmap = hz.getMap(NearCacheTest.MAP);
		testmap.set("3", "3");

		// testmap.put("1", String.valueOf(System.currentTimeMillis()));
		// testmap.put("2", String.valueOf(System.currentTimeMillis()));

		// new NearCacheTest();
		SpringCache.main(args);
		SpringCache.main(args);

		Thread.sleep(3000);
		new Thread(tx, "transaction").start();

		// Hazelcast.shutdownAll();
		// System.exit(0);
	}

	// TODO 实现一个自定义的分布式事务，更新各个client的缓存
	public static void tx() {
		// begin tx
		// lock map entry : JUST USE IMAP::put
		// ACK: clear all the client's cache strongly
		// update/add/delete map entry
		// commit tx
	}
}
